package com.example.mq.server.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.server.VirtualHost;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;


/**
 * 通过这个类来实现消费消息的核心逻辑
 */
@Slf4j
public class ConsumerManager {
    // 持有上层的 VirtualHost 对象 用于操作数据
    private VirtualHost parent;
    // 线程池, 用于执行回调任务
    private final ExecutorService workPoll = Executors.newFixedThreadPool(4);
    // 持有令牌的阻塞队列
    private final BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    // 扫描线程
    private Thread scannerThread = null;


    public ConsumerManager(VirtualHost p){
        this.parent = p;

        scannerThread = new Thread(){
            @Override
            public void run() {
                while(true){
                    try {
                        // 1.拿到令牌
                        String queueName = tokenQueue.take();
                        // 2.根据令牌找到队列
                        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                        if(queue == null){
                            throw new MqException("[ConsumerManager] 取令牌后,发现队列名不存在! queueName = "+queueName);
                        }
                        // 3.从队列中消费一个消息
                        synchronized (queue){
                            consumeMessage(queue);
                        }
                    } catch (InterruptedException | MqException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        };
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    /**
     * 有线程入队列,可以通知消费者进行消费了
     * @param queueName
     * @throws InterruptedException
     */
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if(queue == null){
            throw new MqException("[ConsumerManager] 队列不存在! queueName = "+queueName);
        }

        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue){
            queue.addConsumer(consumerEnv);
            // 如果当前队列已经有了一些消息,需要立即消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for(int i = 0; i<n; i++){
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        // 按照轮询的方式找个消费者出来
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if(luckyDog == null){
            // 当前队列没有消费者,暂时不消费
            return;
        }
        // 2.取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue);
        if(message == null){
            // 当前队列中没有消息,也不需要消费
            return;
        }
        // 3.把消息带入到消费者的回调方法中,丢给线程池执行
        workPoll.submit(()-> {
            try {
                // 1. 把消息放入待确认的消息集合中
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 2.真正执行回调操作
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),
                        message.getBasicProperties(), message.getBody());
                // 如果是自动应答可以直接删除了,如果是手动应答,暂时不处理交给后面的basicAck处理
                if(luckyDog.isAutoAck()){
                    // 1.删除硬盘上的消息
                    if(message.getDeliverMode() == 2){
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    // 2.删除待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(),message.getMessageId());
                    // 3.删除内存中消息中心中的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            log.info("[ConsumerManager] 消息被消费成功! queueName = "+queue.getName());
        });
    }
}
